package com.itbaizhan.rocketmqproducer.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeUnit;
/*异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应*/
//发送异步消息生产者（不等待broker有消息返回，就进行下步工作，可靠性低，效率快）
/*1，创建生产者 2，创建消息 3，异步发送,实现SendCallback()接口*/
public class AsyncProducer {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException {
        //实例化生产者
        DefaultMQProducer producer = new DefaultMQProducer("producer_grp_01");
        //设置nameserver 地址
        producer.setNamesrvAddr("192.168.66.100:9876");
        producer.start();
        //设置发送消息失败时重试次数  0是不重新发送
        producer.setRetryTimesWhenSendAsyncFailed(0);

        //消息的数量
        int messageCount = 100;
        //根据消息总数量实例化倒计时器
        CountDownLatch2 countDownLatch2 = new CountDownLatch2(messageCount);
        for (int i = 0; i < messageCount; i++) {
            final int index = i;
            Message message = new Message("topic_springboot_01","TagA",
                    "OrderID188",
                    ("hello async rockmq .....").getBytes(RemotingHelper.DEFAULT_CHARSET));
            //异步发送消息
            producer.send(message, new SendCallback() {
                //成功发送
                @Override
                public void onSuccess(SendResult sendResult) {
                    //倒计时
                    countDownLatch2.countDown();
                    System.out.printf("%-10d OK %s\n",index,sendResult.getMsgId());
                }
                //失败
                @Override
                public void onException(Throwable throwable) {
                    countDownLatch2.countDown();
                    System.out.printf("%-10d Exception %s\n",index,throwable);
                    throwable.printStackTrace();
                }
            });
        }
        //发送完消息等待5秒，关闭producer
        countDownLatch2.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }
}
